package com.roamer.javase.demo.fm;

import lombok.Builder;
import lombok.Data;

import java.util.*;
import java.util.concurrent.*;

/**
 * @author roamer
 * @version v1.0
 * @since 2022/6/17 01:39
 */
public class NewPush {

    public static void main(String[] args) {
        BlockingQueue<OrderEvent> orderEventQueue = new LinkedBlockingQueue<>(100);
        ExecutorService executorService = new ThreadPoolExecutor(5, 5, 100, TimeUnit.SECONDS, new LinkedBlockingQueue<>(10), new ThreadPoolExecutor.CallerRunsPolicy());
        new Thread(() -> {
            while (true) {
                executorService.execute(() -> {
                    try {
                        String orderCode = "0-" + UUID.randomUUID();
                        int sum = new Random().nextInt(7);
                        for (int i = 0; i < sum; i++) {
                            orderEventQueue.put(OrderEvent.builder().orderCode(orderCode).state(i + 1).build());
                        }
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                });
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }).start();

        Map<String, BlockingQueue<Msg>> queueMap = new HashMap<>(5);
        for (int i = 0; i < 5; i++) {
            queueMap.put(i + "", new LinkedBlockingQueue<>(100));
        }

        Map<String, List<Integer>> versionMap = new ConcurrentHashMap<>();
        ExecutorService executorServiceConsumer = new ThreadPoolExecutor(5, 5, 100, TimeUnit.SECONDS, new LinkedBlockingQueue<>(10), new ThreadPoolExecutor.CallerRunsPolicy());
        new Thread(() -> {
            while (true) {
                executorServiceConsumer.execute(() -> {
                    try {
                        OrderEvent event = orderEventQueue.take();
                        BlockingQueue<Msg> msgQueue = queueMap.get(Math.abs(event.getOrderCode().hashCode()) % 5 + "");
                        Order order = new Order();
                        order.setOrderCode(event.getOrderCode());
                        order.setState(event.getState());
                        // 获取版本号
                        List<Integer> versions = versionMap.computeIfAbsent(event.getOrderCode(), (k) -> new CopyOnWriteArrayList<>());
                        int version;
                        synchronized (versions) {
                            version = versions.isEmpty() ? 1 : versions.get(versions.size() - 1) + 1;
                            versions.add(version);
                        }

                        Msg msg = Msg.builder()
                                .msgId(UUID.randomUUID().toString())
                                .body(order)
                                .version(version)
                                .build();
                        msgQueue.put(msg);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                });
            }
        }).start();
        for (int i = 0; i < 5; i++) {
            BlockingQueue<Msg> msgQueue = queueMap.get(i + "");
            new Thread(() -> {
                while (true) {
                    try {
                        Msg msg = msgQueue.take();
                        Order order = msg.getBody();
                        List<Integer> versions = versionMap.get(order.getOrderCode());
                        if (msg.getVersion() != versions.get(0)) {
                            System.out.printf("还未轮到，重新投递！%s\n", msg);
                            msgQueue.put(msg);
                            continue;
                        }
                        synchronized (versions) {
                            versions.remove(0);
                        }
                        System.out.printf("消费到了！%s\n", msg);

                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
            }).start();

        }

    }


}

@Builder
@Data
class OrderEvent {
    private String orderCode;
    private Integer state;
}

@Builder
@Data
class Msg {
    private String msgId;
    private int version;
    private Order body;
}

@Data
class Order {
    private String orderCode;
    private Integer state;
}
